package com.liujun.io.nio.mysql.protocol;

import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.util.HashMap;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.liujun.io.nio.mysql.bean.SocketDataTran;
import com.liujun.io.nio.mysql.bean.protocol.AuthPackageBean;
import com.liujun.io.nio.mysql.bean.protocol.HandshakeBean;
import com.liujun.io.nio.mysql.bean.protocol.OkPackageBean;
import com.liujun.io.nio.mysql.bean.protocol.QueryPackageBean;
import com.liujun.io.nio.mysql.bean.protocol.result.ColumnPackageBean;
import com.liujun.io.nio.mysql.bean.protocol.result.EofPackageBean;
import com.liujun.io.nio.mysql.bean.protocol.result.ResultSetHanderBean;
import com.liujun.io.nio.mysql.bean.protocol.result.RowDataPackageBean;
import com.liujun.io.nio.mysql.protocol.inf.DataParseInf;
import com.liujun.io.nio.mysql.trans.DataParseProcess;
import com.liujun.io.nio.mysql.util.StringUtil;

/**
 * mysql的协议解析操作
* 源文件名：MysqlProtocolProcess.java
* 文件版本：1.0.0
* 创建作者：Think
* 创建日期：2016年12月9日
* 修改作者：Think
* 修改日期：2016年12月9日
* 文件描述：TODO
* 版权所有：Copyright 2016 zjhz, Inc. All Rights Reserved.
*/
public class MysqlProtocolProcess implements Runnable {

    public static Logger LOGGER = LoggerFactory.getLogger(MysqlProtocolProcess.class);

    /**
    * @字段说明 process
    */
    private final DataParseProcess<SocketDataTran> process;

    /**
     * 向服务器写入数据
    * @字段说明 clientToMysqlServer
    */
    private final DataParseProcess<SocketDataTran> clientToMysqlServer;

    /**
     * 用于进行mysql服务器向中间件发送的握手包
    * @字段说明 serverHandTomid
    */
    private DataParseInf<HandshakeBean> serverHandTomid = new HandshakeBean();

    /**
     * 发送认证操作
    * @字段说明 authProc
    */
    private DataParseInf<AuthPackageBean> authProc = new AuthPackageBean();

    /**
     * 读取成功结果
    * @字段说明 rspPackage
    */
    private DataParseInf<OkPackageBean> rspPackage = new OkPackageBean();

    /**
     * 查询发送的报文
    * @字段说明 queryPack
    */
    private DataParseInf<QueryPackageBean> queryPack = new QueryPackageBean();

    public MysqlProtocolProcess(DataParseProcess<SocketDataTran> process,
            DataParseProcess<SocketDataTran> clientToMysqlServer) {
        this.process = process;
        this.clientToMysqlServer = clientToMysqlServer;
    }

    private int index = 0;

    @Override
    public void run() {

        while (true) {
            // 进行第一次的读取，三握手协议
            SocketDataTran dataBean = process.getValue();

            if (null != dataBean) {

                // HandshakeBean handshake = new HandshakeBean();
                ByteBuffer mapBuf = dataBean.getBuffer();
                int offset = 0;
                // 1获取长度信息, 占3位
                int length = getLength(mapBuf, offset);
                // 2,获取到通讯序号,第四位
                int seq = mapBuf.get(3);
                System.out.println("当前mysql通讯序号:" + seq);

                final String hexs = StringUtil.dumpAsHex(mapBuf, 0, length);

                System.out.println("当前第" + index + "响应");

                LOGGER.info("C#{}B#{}: last readed = {} bytes, total readed = {} bytes, buffer bytes\n{}",
                        Thread.currentThread().getId(), mapBuf.hashCode(), 0, length, hexs);

                // //
                // 3,数据具体内容,在第一次通讯的时候，需要解读响应的报文类型信息,首次为10，####,在首次是服务端面服务协议版本号
                int msgType = mapBuf.get(4);
                System.out.println("消息的类型信息:" + msgType);

                if (index == 0) {
                    HadnShakeprocSocket(dataBean, length);
                }
                // 进行第一次服务器端的响应操作,然后发送查询 SQL
                else if (index == 1) {
                    sendQueryStrSocket(dataBean, length);
                }
                // 进行查询结果解码
                else if (index == 2) {

                    ResultSetRead read = new ResultSetRead();

                    int oldPos = mapBuf.position();
                    mapBuf.position(0);
                    ByteBuffer copyBuf = mapBuf.slice();
                    copyBuf.limit(mapBuf.limit());
                    mapBuf.position(oldPos);

                    read.readResultset(copyBuf);

                    // queryResponseHandle(copyBuf);
                }

                index++;
            }
        }
    }

    private void queryResponseHandle(ByteBuffer mapBuf) {

        int oldPos = mapBuf.position();
        mapBuf.position(0);
        ByteBuffer copyBuf = mapBuf.slice();
        copyBuf.limit(mapBuf.limit());
        mapBuf.position(oldPos);

        // ,1[Result Set Header] 列数量
        DataParseInf<ResultSetHanderBean> resultDataColumnNum = new ResultSetHanderBean();

        ResultSetHanderBean columnNumPackage = resultDataColumnNum.readPackage(copyBuf);

        // 2,循环列信息
        DataParseInf<ColumnPackageBean> columnDataProcess = null;
        for (int i = 0; i < columnNumPackage.getFieldCount(); i++) {
            columnDataProcess = new ColumnPackageBean();
            // 进行循环列信息
            ColumnPackageBean column = columnDataProcess.readPackage(copyBuf);
            System.out.println("读取的列的数据：" + column);
        }

        // 3,列结束
        DataParseInf<EofPackageBean> eofPackageProcess = new EofPackageBean();
        eofPackageProcess.readPackage(copyBuf);

        // 4，进行行数据读取
        DataParseInf<RowDataPackageBean> rowDataProcess = new RowDataPackageBean(columnNumPackage.getFieldCount());

        // 读取报文
        while (!checkEofAackage(copyBuf)) {
            RowDataPackageBean rowData = rowDataProcess.readPackage(copyBuf);
            System.out.println("读取的行数据：" + rowData);
        }

        // 读取最后的eof报文
        eofPackageProcess.readPackage(copyBuf);
        // 5,一直读取到EOF报文为止
    }

    /**
     * 检查是否为eof的标识
    * 方法描述
    * @param copyBuf
    * @return
    * @创建日期 2016年12月14日
    */
    public boolean checkEofAackage(ByteBuffer copyBuf) {
        // 以当前的position坐标标，向后读取4个
        int currPositon = copyBuf.position();
        // 读取新坐标
        if (copyBuf.limit() >= currPositon + 4) {
            copyBuf.position(currPositon + 4);
            byte bufValue = copyBuf.get();
            if (bufValue == EofPackageBean.DEFVALUE) {
                // 重新设置为老坐标
                copyBuf.position(currPositon);
                return true;
            }
        }

        return false;
    }

    /**
     * 进行数据的握手写入
     * 方法描述
     * @param dataBean
     * @param length
     * @创建日期 2016年12月13日
     */
    private void sendQueryStrSocket(SocketDataTran dataBean, int length) {
        ByteBuffer mapBuf = dataBean.getBuffer();

        int oldPos = mapBuf.position();
        mapBuf.position(0);
        ByteBuffer copyBuf = mapBuf.slice();
        copyBuf.limit(length);
        mapBuf.position(oldPos);

        OkPackageBean okPackss = rspPackage.readPackage(copyBuf);

        System.out.println("最终的结果:" + okPackss);

        // 组装查询报文
        ByteBuffer buffer = queryPack.getWriteToPackage(null);

        System.out.println("查询组装报文:");

        dataBean.getBuffer().clear();

        SocketDataTran queryBean = new SocketDataTran(buffer, dataBean.getSelectKey());
        // 在本身的键中注册一个写入事件
        queryBean.getSelectKey().interestOps(dataBean.getSelectKey().interestOps() | SelectionKey.OP_WRITE);
        // 将当前查询请求放入队列
        clientToMysqlServer.putValue(queryBean);

    }

    /**
     * 首次数据写入
    * 方法描述
    * @param dataBean
    * @param length
    * @创建日期 2016年12月13日
    */
    private void HadnShakeprocSocket(SocketDataTran dataBean, int length) {
        ByteBuffer mapBuf = dataBean.getBuffer();

        int oldPos = mapBuf.position();
        mapBuf.position(0);
        ByteBuffer copyBuf = mapBuf.slice();
        copyBuf.limit(length);
        mapBuf.position(oldPos);

        // 1, 首次从mysql服务端到中间件的握手包读取
        HandshakeBean hand = serverHandTomid.readPackage(copyBuf);

        Map<String, Object> param = new HashMap<>();
        param.put("handShakeBean", hand);

        // 2, 发送用户名及密码
        ByteBuffer authPackage = authProc.getWriteToPackage(param);

        dataBean.getBuffer().clear();

        SocketDataTran dataTran = new SocketDataTran(authPackage, dataBean.getSelectKey());
        // 在本身的键中注册一个写入事件
        dataBean.getSelectKey().interestOps(dataBean.getSelectKey().interestOps() | SelectionKey.OP_WRITE);
        // 注册写入事件
        clientToMysqlServer.putValue(dataTran);
    }

    public int getLength(ByteBuffer buffer, int offset) {
        // mysql包头的长度
        int msyql_packetHeaderSize = 4;
        int length = buffer.get(offset) & 0xff;
        length |= (buffer.get(++offset) & 0xff) << 8;
        length |= (buffer.get(++offset) & 0xff) << 16;
        return length + msyql_packetHeaderSize;
    }

}
